package sqlengine.queryExecution;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;
import java.util.Vector;

import sqlengine.conditions.JoinCondition;
import sqlengine.conditions.NonJoinCondition;
import sqlengine.database.DataFile;
import sqlengine.database.DataManager;
import sqlengine.database.Index;
import sqlengine.database.Index.RecordArray;
import sqlengine.database.impl.Record;

public class BasicOperation {

	static ArrayList<Record> tableScan(String tableName) {
		Map<String, DataFile> files = DataManager.getOpenedFiles();
		if (!(files.containsKey(tableName))) {
			throw new IllegalArgumentException("table does not exist!");
		}
		DataFile df = files.get(tableName);

		ArrayList<Record> result = new ArrayList<Record>();
		Iterator<Integer> it = df.iterator();

		while (it.hasNext()) {
			result.add(df.getrecord(it.next()));
		}
		return result;
	}

	static ArrayList<Record> indexScan(NonJoinCondition nonJoinCondition)
			throws BasicOperationException {
		String tableName;
		String columnName;
		String value;

		tableName = nonJoinCondition.getTableName();
		columnName = nonJoinCondition.getColumnName();
		value = nonJoinCondition.getValue();

		Map<String, DataFile> files = DataManager.getOpenedFiles();
		if (!(files.containsKey(tableName))) {
			throw new IllegalArgumentException("table does not exist!");
		}
		DataFile df = files.get(tableName);

		ArrayList<Record> result = new ArrayList<Record>();

		Index index = df.getIndex(columnName);
		if (index == null) {
			throw new BasicOperationException("can't find index on column"
					+ columnName);
		}
		Iterator<Integer> it = index.iterator(value);
		while (it.hasNext()) {
			result.add(df.getrecord(it.next()));
		}

		return result;
	}

	static ArrayList<Record> cartesian_product(ArrayList<Record> records1,
			ArrayList<Record> records2) {
		// don't change the record schema, just add it up, the tableNames order
		// is consistent with tableNames variable
		ArrayList<Record> result = new ArrayList<Record>();
		Iterator<Record> it1 = records1.iterator();
		while (it1.hasNext()) {
			Record firstOne = it1.next();
			Iterator<Record> it2 = records2.iterator();
			while (it2.hasNext()) {
				Record secondOne = it2.next();
				result.add(firstOne);
				result.add(secondOne);
			}
		}
		return result;
	}

	static ArrayList<Record> nested_loops_join(ArrayList<Record> records1,
			ArrayList<Record> records2, JoinCondition joinCondition) {
		// don't change the record schema, just add it up, the tableNames order
		// is consistent with tableNames variable
		ArrayList<Record> result = new ArrayList<Record>();
		Iterator<Record> it1 = records1.iterator();
		while (it1.hasNext()) {
			Record firstOne = it1.next();
			Iterator<Record> it2 = records2.iterator();

			while (it2.hasNext()) {
				Record secondOne = it2.next();
				if (firstOne.getKey(joinCondition.getColumnName1()).compareTo(
						secondOne.getKey(joinCondition.getColumnName2())) == 0) {
					result.add(firstOne);
					result.add(secondOne);
				}
			}
		}
		return result;
	}

	// for one column has index
	static ArrayList<Record> index_join(ArrayList<Record> records,
			JoinCondition joinCondition, String[] tableNames)
			throws BasicOperationException {
		// assume the first condition has index
		// don't change the record schema, just add it up, the tableNames order
		// is consistent with tableNames variable
		ArrayList<Record> result = new ArrayList<Record>();
		Iterator<Record> it = records.iterator();

		while (it.hasNext()) {
			Record firstRecord = it.next();
			String key = firstRecord.getKey(joinCondition.getColumnName2());
			ArrayList<Record> secondRecords = indexScan(new NonJoinCondition(
					joinCondition.getTableName1(),
					joinCondition.getColumnName1(), key));
			Iterator<Record> it2 = secondRecords.iterator();
			while (it2.hasNext()) {
				if (joinCondition.getTableName1().compareTo(tableNames[0]) == 0) {
					result.add(it2.next());
					result.add(firstRecord);
				} else {
					result.add(firstRecord);
					result.add(it2.next());
				}
			}
		}
		return result;
	}

	// for two column both have index
	static ArrayList<Record> index_join_both(JoinCondition joinCondition)
			throws BasicOperationException {
		// don't change the record schema, just add it up, the tableNames order
		// is consistent with tableNames variable
		ArrayList<Record> result = new ArrayList<Record>();
		DataFile dataFile1 = DataManager.getOpenedFiles().get(
				joinCondition.getTableName1());
		DataFile dataFile2 = DataManager.getOpenedFiles().get(
				joinCondition.getTableName2());
		Index index1 = dataFile1.getIndex(joinCondition.getColumnName1());
		if (index1 == null) {
			throw new BasicOperationException("can't find index on column"
					+ joinCondition.getColumnName1());
		}
		Index index2 = dataFile2.getIndex(joinCondition.getColumnName2());
		if (index2 == null) {
			throw new BasicOperationException("can't find index on column"
					+ joinCondition.getColumnName2());
		}
		Iterator<RecordArray> it1 = index1.getBPlusTreeiterator();
		Iterator<RecordArray> it2 = index2.getBPlusTreeiterator();
		if (it1.hasNext() && it2.hasNext()) {
			RecordArray recordArray1 = it1.next();
			RecordArray recordArray2 = it2.next();
			do {
				if (recordArray1.getKey().compareTo(recordArray2.getKey()) == 0) {
					for (int i = 0; i < recordArray1.size(); i++) {
						for (int j = 0; j < recordArray2.size(); j++) {
							result.add(dataFile1.getrecord(recordArray1.get(i)));
							result.add(dataFile2.getrecord(recordArray2.get(j)));
						}
					}
					if (it1.hasNext() && it2.hasNext()) {
						recordArray1 = it1.next();
						recordArray2 = it2.next();
					} else {
						break;
					}
				} else if (recordArray1.getKey().compareTo(
						recordArray2.getKey()) > 0) {
					if (it2.hasNext()) {
						recordArray2 = it2.next();
					} else {
						break;
					}
				} else {
					if (it1.hasNext()) {
						recordArray1 = it1.next();
					} else {
						break;
					}
				}
			} while (true);
		}
		return result;
	}

	static ArrayList<Record> filter(ArrayList<Record> records,
			ArrayList<JoinCondition> joinConditions,
			ArrayList<NonJoinCondition> nonJoinConditions, String tableNames[]) {
		// don't change the record schema, just add it up, the tableNames order
		// is consistent with tableNames variable
		ArrayList<Record> result = new ArrayList<Record>();
		Iterator<Record> it = records.iterator();
		if (tableNames.length == 1) {
			// one table
			while (it.hasNext()) {
				Record record = it.next();
				boolean failed = false;
				// check join condition
				Iterator<JoinCondition> itjoin = joinConditions.iterator();
				while (itjoin.hasNext()) {
					JoinCondition join = itjoin.next();
					if (record.getKey(join.getColumnName1()).compareTo(
							record.getKey(join.getColumnName2())) != 0) {
						failed = true;
						break;
					}
				}
				// check nonjoin condition
				Iterator<NonJoinCondition> itNonJoin = nonJoinConditions
						.iterator();
				while (itNonJoin.hasNext()) {
					NonJoinCondition nonjoin = itNonJoin.next();
					if (record.getKey(nonjoin.getColumnName()).compareTo(
							nonjoin.getValue()) != 0) {
						failed = true;
						break;
					}
				}
				if (!failed) {
					result.add(record);
				}
			}
		} else {
			// two tables
			while (it.hasNext()) {
				Record record1 = it.next();
				Record record2 = it.next();
				boolean failed = false;
				// check join condition
				Iterator<JoinCondition> itjoin = joinConditions.iterator();
				while (itjoin.hasNext()) {
					JoinCondition join = itjoin.next();
					Record lhsRecord, rhsRecord;
					if (join.getTableName1().compareTo(tableNames[0]) == 0) {
						lhsRecord = record1;
					} else {
						lhsRecord = record2;
					}
					if (join.getTableName2().compareTo(tableNames[0]) == 0) {
						rhsRecord = record1;
					} else {
						rhsRecord = record2;
					}
					if (lhsRecord.getKey(join.getColumnName1()).compareTo(
							rhsRecord.getKey(join.getColumnName2())) != 0) {
						failed = true;
						break;
					}
				}
				// check nonjoin condition
				Iterator<NonJoinCondition> itNonJoin = nonJoinConditions
						.iterator();
				while (itNonJoin.hasNext()) {
					NonJoinCondition nonjoin = itNonJoin.next();
					if (nonjoin.getTableName().compareTo(tableNames[0]) == 0) {
						if (record1.getKey(nonjoin.getColumnName()).compareTo(
								nonjoin.getValue()) != 0) {
							failed = true;
							break;
						}
					} else {
						if (record2.getKey(nonjoin.getColumnName()).compareTo(
								nonjoin.getValue()) != 0) {
							failed = true;
							break;
						}
					}
				}
				if (!failed) {
					result.add(record1);
					result.add(record2);
				}
			}
		}
		return result;
	}
}
